/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.rocketmq;

import cn.ac.iie.di.global.GlobalParas;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import org.apache.log4j.Logger;

/**
 *
 * @author YG
 */
public class RocketMqClient implements Runnable {

    public static Logger log = Logger.getLogger(RocketMqClient.class.getName());
    public String oriMqString = "";
    public String oriMq = "";
    public String group = "";

    private REReceiveSession session;

    public RocketMqClient(String oriMqString) {
        this.oriMq = oriMqString.split("[:]")[0];
        this.group = oriMqString.split("[:]")[1];
    }

    public REReceiveSession getSession() {
        return session;
    }

    @Override
    public void run() {
        //生成连接，参数为rocketmq的nameserver串。可以多个nameserver，分号隔开。由具体业务决定
        REConnection conn = new REConnection(GlobalParas.nameServer);
        //生成builder，参数为topic，由具体业务决定
        REReceiveSessionBuilder builder = null;
        try {
            builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder(this.oriMq);
        } catch (REConnectionException ex) {
            log.error(ex, ex);
        }
        //设置receiver 的两个处理函数：
        //一个是正常数据处理，实现REAbstractReceiveMessageHandler<REMessage>接口。返回true则成功，否则会发给下一个handler。
        if (builder == null) {
            log.error("init receiver fail , will exit!");
            System.exit(0);
        }
        builder.setHandler(new REAbstractReceiveMessageHandler<REMessage>() {
            @Override
            public boolean handle(REMessage message) {
                if (message == null || message.data.size() == 0) {
                    return false;
                }
                try {
                    //和put一样，但是不会像put那样阻塞住，如果put不进去，可以在下面报错
                    boolean offer = GlobalParas.oriMsgPool.offer(message);
//                                        log.info("get " + message.data.size() + " message from " + oriMq);
                    if (!offer) {
                        //TODO 增加日志
                        while (true) {
                            offer = GlobalParas.oriMsgPool.offer(message);
                            if (offer == true) {
                                log.info("get " + message.data.size() + " message from " + oriMq);
                                break;
                            }
                        }
                    } else {
                        log.info("get " + message.data.size() + " message from " + oriMq);
                    }
                } catch (Exception ex) {
                    log.error(ex, ex);
                }
                return true;
            }
        });
        //另一个是错误数据处理，实现REAbstractReceiveMessageHandler<byte[]>接口。返回true则成功，否则会退回去，重新处理。
        builder.setFailureHandler(new REAbstractReceiveMessageHandler<byte[]>() {
            @Override
            public boolean handle(byte[] message) {
                System.out.println("formatter fuilure ! size = " + message.length);
                return true;
            }
        });
        //设置组名。相同的组名处理的是同一份数据，不同的组名处理的是不同的数据备份。
        builder.setGroupName(this.group);
        //设置从处理起始位置。如果设置的是CONSUME_FROM_TIMESTAMP，需要同时设置起始时间戳
        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
//        builder.setConsumeTimestamp(new Date().getTime());
//        System.out.println(new Date().toLocaleString());
        //设置消费线程数
        builder.setConsumeThreadNum(8);
        //生成session
//        REReceiveSession session;
        try {
            session = (REReceiveSession) builder.build();
            session.start();
        } catch (REConnectionException ex) {
        }
    }

    public void shutDownConsumer() throws InterruptedException {
        if (!GlobalParas.pullMsgQueue.isEmpty()) {
            try {
                session.shutdown();
            } catch (REConnectionException ex) {
                log.error("shut down session err!!!");
            }
        }

    }
}
